package cn.itcast.hudi.edu;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 基于Flink CDC Connector实现：实时获取MySQL数据库中传智教育业务数据，转换处理后，实时存储Hudi表
 */
public class EduDataStreamHudi {

	public static void main(String[] args) {
		// 1-获取表执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(2);
		env.enableCheckpointing(5000);
		EnvironmentSettings settings = EnvironmentSettings
			.newInstance()
			.inStreamingMode()
			.build();
		
//		流式的表执行环境
		StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings) ;

		// 2-基于CDC方式，将数据库MySQL中表数据实时增量同步到Hudi表

	}

	/**
	 * 具体实现CDC 操作，基于Flink SQL完成
	 */
	public static void cdcMySQLToHudi(StreamTableEnvironment tableEnv){
		// step1、创建输入表 - TODO: CDC
		tableEnv.executeSql("   create table tbl_customer_mysql (\r\n" + 
				"  id STRING PRIMARY KEY NOT ENFORCED,\r\n" + 
				"  customer_relationship_id STRING,\r\n" + 
				"  create_date_time STRING,\r\n" + 
				"  update_date_time STRING,\r\n" + 
				"  deleted STRING,\r\n" + 
				"  name STRING,\r\n" + 
				"  idcard STRING,\r\n" + 
				"  birth_year STRING,\r\n" + 
				"  gender STRING,\r\n" + 
				"  phone STRING,\r\n" + 
				"  wechat STRING,\r\n" + 
				"  qq STRING,\r\n" + 
				"  email STRING,\r\n" + 
				"  area STRING,\r\n" + 
				"  leave_school_date STRING,\r\n" + 
				"  graduation_date STRING,\r\n" + 
				"  bxg_student_id STRING,\r\n" + 
				"  creator STRING,\r\n" + 
				"  origin_type STRING,\r\n" + 
				"  origin_channel STRING,\r\n" + 
				"  tenant STRING,\r\n" + 
				"  md_id STRING\r\n" + 
				")WITH (\r\n" + 
				"  'connector' = 'mysql-cdc',\r\n" + 
				"  'hostname' = '192.168.249.112',\r\n" + 
				"  'port' = '3306',\r\n" + 
				"  'username' = 'root',\r\n" + 
				"  'password' = 'root',\r\n" + 
				"  'server-time-zone' = 'Asia/Shanghai',\r\n" + 
				"  'debezium.snapshot.mode' = 'initial',\r\n" + 
				"  'database-name' = 'itcast_nev',\r\n" + 
				"  'table-name' = 'customer'\r\n" + 
				"); ");

		// step2、创建输出表
		tableEnv.executeSql(" CREATE TABLE edu_customer_hudi(\r\n" + 
				"  id STRING PRIMARY KEY NOT ENFORCED,\r\n" + 
				"  customer_relationship_id STRING,\r\n" + 
				"  create_date_time STRING,\r\n" + 
				"  update_date_time STRING,\r\n" + 
				"  deleted STRING,\r\n" + 
				"  name STRING,\r\n" + 
				"  idcard STRING,\r\n" + 
				"  birth_year STRING,\r\n" + 
				"  gender STRING,\r\n" + 
				"  phone STRING,\r\n" + 
				"  wechat STRING,\r\n" + 
				"  qq STRING,\r\n" + 
				"  email STRING,\r\n" + 
				"  area STRING,\r\n" + 
				"  leave_school_date STRING,\r\n" + 
				"  graduation_date STRING,\r\n" + 
				"  bxg_student_id STRING,\r\n" + 
				"  creator STRING,\r\n" + 
				"  origin_type STRING,\r\n" + 
				"  origin_channel STRING,\r\n" + 
				"  tenant STRING,\r\n" + 
				"  md_id STRING,\r\n" + 
				"  part STRING\r\n" + 
				")\r\n" + 
				"PARTITIONED BY (part)\r\n" + 
				"WITH(\r\n" + 
				"  'connector'='hudi',\r\n" + 
				"  'path'= 'hdfs://192.168.249.113:9000/hudi-warehouse/edu_customer_hudi', \r\n" + 
				"  'table.type'= 'MERGE_ON_READ',\r\n" + 
				"  'hoodie.datasource.write.recordkey.field'= 'id', \r\n" + 
				"  'write.precombine.field'= 'create_date_time',\r\n" + 
				"  'write.tasks'= '1',\r\n" + 
				"  'write.rate.limit'= '2000', \r\n" + 
				"  'compaction.tasks'= '1', \r\n" + 
				"  'compaction.async.enabled'= 'true',\r\n" + 
				"  'compaction.trigger.strategy'= 'num_commits',\r\n" + 
				"  'compaction.delta_commits'= '1',\r\n" + 
				"  'changelog.enabled'= 'true'\r\n" + 
				");\r\n" + 
				"") ;

		// step3、查询输入表数据，并且可以进行处理操作，最后插入到输出表中
		tableEnv.executeSql(" insert into edu_customer_hudi select *, CAST(CURRENT_DATE AS STRING) AS part from tbl_customer_mysql ") ;
	}

}

